"""
@-*- coding: utf-8 -*-
@ python：python 3.8
@ 创建人员：ylx
@ 创建时间：2025/4/8 下午1:53
@ 内容：
"""
import subprocess
import wexpect
import time
import json
import re
import allure


# 基础工具模块
# 提供通过 cmd.exe 执行相关命令的方法


class CMDSession:
    """
    功能：
    使用subprocess来模拟cmd.exe命令，保证上下文会话一致性

    使用限制：
    非交互式命令可以使用此方法

    举例：
    先cd到 d:/
    再执行 armory init @namespace/package@1.0.0
    预期 在d盘目录下生成对应的armory 文件夹

    此方法不能执行交互式命令，仅用于保持上下文会话一致性
    """

    def __init__(self):
        # 启动一个长期运行的 cmd.exe 进程
        # 将标准错误和标准输出合并到同一个流中
        # 解决armory login -u yanglingxiao -p 缺少参数时，客户端报错展示
        # 原因，系统的标准错误流之前的重定向流的代码未做细分，导致仅重定向了标准输出流
        # 从而导致错误原因无法看到
        # 方案：合并至一个流，错误信息也是执行测试过程中需要的
        self.process = subprocess.Popen(
            ['cmd.exe'],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT
        )

        # 跳过启动cmd时开头的相关信息，如Microsoft Windows [版本 10.0.26100.3194]等无用信息
        # 使用while循环，开始时，逐行读取流的输出，读取到空行时结束循序，跳过开头
        while True:
            line = self.process.stdout.readline().decode('gbk').encode('utf-8').decode('utf-8')
            if line.strip() == "":
                break

    def execute(self, command):
        """
        模拟在 cmd.exe 中执行命令并打印stdout的通用方法
        :param command:具体执行命令
        :return:
        """
        # 发送命令到 cmd.exe 并读取输出
        self.process.stdin.write((command + '\n').encode('gbk'))
        # 强制推送输入，确保命令执行
        self.process.stdin.flush()

        # 给 allure 报告增加细节-测试命令
        allure.attach(f"{command}", "armory 命令", attachment_type=allure.attachment_type.TEXT)

        output_info = []
        nn = 0
        # 逐行读取管道输出，遇到匹配的命令行时，停止当前循环
        # 目的：读取输入的命令行的回显信息
        while True:
            line = self.process.stdout.readline().decode('gbk').encode('utf-8').decode('utf-8')
            # 若开始就读取到空行，则跳过空行，连续三次读取到空行，第4次时结束循环
            if line.strip() == "":
                nn += 1
                if nn > 3:
                    break
                continue

            # 打印流中的输出，预期此处应该打印的是相关命令的回显
            print(line, end='')
            output_info.append(line.strip())

            # 若输入的命令在该回显内，则停止当前循环
            if f"{command}" in line.strip():
                break

        empty_line_count = 0
        timeout = 10
        start_time = time.time()

        # 读取命令执行后的数据,结合项目实际情况会有1次空行打印，然后才有输出
        # 故仅读取一次空行,目的是为读取空行后的值
        # 连续空行则结束循环
        # 后续空行可不读取，因为下一个命令开始后，前述的空行均会被跳过
        while True:
            line = self.process.stdout.readline().decode('utf-8')
            if line.strip() == "":
                empty_line_count += 1
                if empty_line_count > 1:
                    break
                continue
            else:
                empty_line_count = 0
                print(line.strip())
                output_info.append(line.strip())

        allure.attach(f"{output_info}", "响应结果", attachment_type=allure.attachment_type.JSON)

        return output_info

    def execute_1(self, command):
        """
        执行命令并打印stdout单一方法
        为了解决执行命令后，仅吐1个空行的情况
        如 cd d:/  或 rm -rf 文件夹  等情况
        既输出命令后，仅吐回一个空行
        可以使用该方法，避免死循环
        :param command: 具体执行的命令，如cd d:/
        :return:
        """
        # 发送命令到 cmd.exe 并读取输出
        self.process.stdin.write((command + '\n').encode('gbk'))
        # 强制推送输入，确保命令执行
        self.process.stdin.flush()

        # 给 allure 报告增加细节-测试命令
        allure.attach(f"{command}", "armory 命令", attachment_type=allure.attachment_type.TEXT)

        output_info = []
        nn = 0
        # 逐行读取管道输出，遇到匹配的命令行时，停止当前循环
        while True:
            line = self.process.stdout.readline().decode('gbk').encode('utf-8').decode('utf-8')
            # 跳过空行
            if line.strip() == "":
                nn += 1
                if nn > 3:
                    print('-----')
                    break
                continue
            print(line, end='')  # 调试信息
            output_info.append(line.strip())
            if f"{command}" in line.strip():  # 检测提示符
                break

        # 读取到命令执行后的数据,
        # 若读取到空行则结束循环
        while True:
            line = self.process.stdout.readline().decode('gbk')

            if line.strip() == "":
                break
            else:
                print(line.strip())
                output_info.append(line.strip())

        allure.attach(f"{output_info}", "响应结果", attachment_type=allure.attachment_type.JSON)

    def close(self):
        # 关闭会话
        self.process.stdin.write('exit\n'.encode('gbk'))
        self.process.stdin.flush()
        self.process.wait()


def run_command(command):
    """
    仅执行单个简单命令的方法
    不保持会话一致性
    不能执行交互式命令
    :param command: 执行命令，如armory whoami
    :return: 执行命令后，全部的标准输出，标准错误
    """

    allure.attach(f"{command}", "armory 命令", attachment_type=allure.attachment_type.TEXT)
    output_info = []
    result = subprocess.run(
        command,
        capture_output=True,
        text=True,
        shell=True
    )
    output_info.append(result.stdout)
    output_info.append(result.stderr)

    allure.attach(f"{output_info}", "响应结果", attachment_type=allure.attachment_type.JSON)
    return result.stdout, result.stderr


def run_interactive_command_login(**kwargs):
    """
    执行交互式：登录的命令方法
    方法整体核心依赖 wexpect 方法，借助捕获相关的数据后，执行命令
    该方法的实现逻辑，可以了解完 wexpect 方法后即可明白
    打印捕获到时及之前的输出
    最后将分段捕获的输出组合起来
    整体输出
    :param kwargs: 命令/用户名/密码
    :return:
    """
    command = kwargs.get("command")
    allure.attach(f"{command}", "armory 命令", attachment_type=allure.attachment_type.TEXT)
    username = kwargs.get("username")
    password = kwargs.get("password")
    child = wexpect.spawn("cmd.exe")

    child.expect(r">", timeout=10)
    child.sendline(command)
    child.expect("Enter your username: ")
    for char in username:
        child.send(char)
        time.sleep(0.1)  # 模拟按键间隔
    child.send("\r")  # 发送回车键

    # 捕获第一次匹配时的相关输出
    aa1 = child.before + child.after

    child.expect("Enter your password: ")
    for char in password:
        child.send(char)
        time.sleep(0.1)  # 模拟按键间隔
    child.send("\r")  # 发送回车键

    # 捕获第二次匹配时的相关输出
    aa2 = child.before + child.after

    try:
        # child.expect(r">", timeout=10)
        # child.sendline(command)
        # child.expect("Enter your username: ")
        # for char in username:
        #     child.send(char)
        #     time.sleep(0.1)  # 模拟按键间隔
        # child.send("\r")  # 发送回车键
        #
        # aa1 = child.before+child.after
        #
        # child.expect("Enter your password: ")
        # for char in password:
        #     child.send(char)
        #     time.sleep(0.1)  # 模拟按键间隔
        # child.send("\r")  # 发送回车键
        #
        # aa2 = child.before+child.after
        # 捕获等待密码输入******，避免打印******
        child.expect(r"\*\*\*\*\*\*", timeout=10)

        # 捕获第三次输出
        aa3 = child.before + child.after

        # 整合上述捕获的全部输出
        aas = aa1 + aa2 + aa3

        # 适配交互式登录的三种情况：成功、输入错误密码，不输入密码
        index = child.expect([r"Login succeed!", "账号将被锁定。"], timeout=10)

        # 整合全部捕获的输出，并将字符串处理为列表
        output_before = aas + child.before + child.after
        output_after = output_before.split("\r\n")

        try:
            while True:
                # 循环删除列表中所有空行
                output_after.remove("")
        except ValueError:
            # 捕获异常后，将所有数据组成字符串，并按相关格式输出
            result = "\r\n".join(output_after)

        if index == 0:
            allure.attach(f"{result}", "响应结果", attachment_type=allure.attachment_type.JSON)
            return result
        elif index == 1:
            allure.attach(f"{result}", "响应结果", attachment_type=allure.attachment_type.JSON)
            return result
            # return child.before + child.after

    except wexpect.TIMEOUT:
        child.expect("密码不能为空")
        aa3 = child.before + child.after
        # return result
        # return child.before + child.after
        output_before = aa1 + aa2 + aa3
        output_after = output_before.split("\r\n")

        try:
            while True:
                output_after.remove("")
        except ValueError:
            result = "\r\n".join(output_after)
            allure.attach(f"{result}", "响应结果", attachment_type=allure.attachment_type.JSON)
            return result

    finally:
        child.sendline('exit')
        child.close()


def run_interactive_command_publish(**kwargs):
    """
    执行交互式：发布软件包的命令方法
    处理了发布过程中，如开启了客户端更新提示后的交互式处理
    :param kwargs: 命令/选择是否更新client，一般设置为n
    :return:
    """
    command = kwargs.get("command")
    allure.attach(f"{command}", "armory 命令", attachment_type=allure.attachment_type.TEXT)
    choose = kwargs.get("choose")
    child = wexpect.spawn("cmd.exe")
    try:
        child.expect([r">"], timeout=10)
        child.sendline(command)

        # 处理是否有升级客户端的提示，若有则按publish_info 中的配置完成输入
        #建议可以发布时，将armory.config.json的配置中，客户端更新提示关闭
        index = child.expect(["n]"], timeout=10)
        if index == 0:
            for char in choose:
                child.send(char)
                time.sleep(0.1)  # 模拟按键间隔
            child.send("\r")  # 发送回车键
            child.expect("Package publish completed!")
            allure.attach(f"{child.before + child.after}", "响应结果", attachment_type=allure.attachment_type.JSON)
            return child.before + child.after
    except wexpect.TIMEOUT:
        #异常处理捕获不到预期文本时的处理
        child.expect("Package publish completed!")
        allure.attach(f"{child.before + child.after}", "响应结果", attachment_type=allure.attachment_type.JSON)
        return child.before + child.after
    finally:
        child.sendline('exit')
        child.close()


def run_interactive_command_download(**kwargs):
    """
    执行交互式：下载软件包命令方法
    若遇到选择版本升级时，可使用此方法
    同上述发布软件包时遇到的场景一致
    因为捕获的文本不同，故另起一个方法
    :param kwargs: 命令/选择是否更新client
    :return:
    """
    command = kwargs.get("command")
    allure.attach(f"{command}", "响应结果", attachment_type=allure.attachment_type.TEXT)
    choose = kwargs.get("choose")
    child = wexpect.spawn("cmd.exe")
    try:
        child.expect([r">"], timeout=10)
        child.sendline(command)
        index = child.expect(["n]"], timeout=10)
        if index == 0:
            for char in choose:
                child.send(char)
                time.sleep(0.1)  # 模拟按键间隔
            child.send("\r")  # 发送回车键
            child.expect("Download is completed!")
            allure.attach(f"{child.before + child.after}", "响应结果", attachment_type=allure.attachment_type.JSON)
            return child.before + child.after
    except wexpect.TIMEOUT:
        child.expect("Download is completed!")
        allure.attach(f"{child.before + child.after}", "响应结果", attachment_type=allure.attachment_type.JSON)
        return child.before + child.after
    finally:
        child.sendline('exit')
        child.close()


def run_interactive_command_repo_1(**kwargs):
    """
    执行交互式：切换仓库方法-选择新仓库
    实现以按下键几次后回车来选择
    建议测试时，以选择最后一个仓库为主
    :param kwargs:
    :return:
    """
    command = kwargs.get("command")
    allure.attach(f"{command}", "armory 命令", attachment_type=allure.attachment_type.TEXT)
    target_options_index = kwargs.get("target_options_index", 0)
    child = wexpect.spawn("cmd.exe")
    try:
        child.expect([r">"], timeout=10)
        child.sendline(command)

        index = child.expect(["Enter to confirm."], timeout=10)
        out_aa = child.before + child.after
        if index == 0:
            for char in range(int(target_options_index)):
                child.send("\x1b[B")
                time.sleep(0.1)  # 模拟按键间隔
            child.send("\r")  # 发送回车键
            child.expect("the new repository.")
            allure.attach(f"{out_aa + child.before + child.after}", "响应结果",
                          attachment_type=allure.attachment_type.JSON)
            return out_aa + child.before + child.after
    except wexpect.TIMEOUT as e:
        print('error', e)
    finally:
        child.sendline('exit')
        child.close()


def run_interactive_command_repo_2(**kwargs):
    """
    执行交互式命令：切换仓库-仍然选择原有仓库
    既进入选择后，不做下键选择，直接回车
    调用2次后，可以实现预期场景
    :param kwargs: 命令/选择是否更新client
    :return:
    """
    command = kwargs.get("command")
    allure.attach(f"{command}", "armory 命令", attachment_type=allure.attachment_type.TEXT)
    target_options_index = kwargs.get("target_options_index", 0)
    child = wexpect.spawn("cmd.exe")
    try:
        child.expect([r">"], timeout=10)
        child.sendline(command)

        index = child.expect(["Enter to confirm."], timeout=10)
        out_aa = child.before + child.after
        if index == 0:
            child.send("\r")  # 发送回车键
            child.expect("is active now")
            allure.attach(f"{out_aa + child.before + child.after}", "响应结果",
                          attachment_type=allure.attachment_type.JSON)
            return out_aa + child.before + child.after
    except wexpect.TIMEOUT as e:
        print('error', e)
    finally:
        child.sendline('exit')
        child.close()


def run_interactive_command_remote_1(**kwargs):
    """
    执行交互式：切换远程仓库 命令方法-选择新仓库
    由于远程仓库的选择新仓库或者重复选择仓库后
    输出的文档有一致的地方，所以使用一个方法即可
    完成2个场景的需要
    :param kwargs:
    :return:
    """
    command = kwargs.get("command")
    allure.attach(f"{command}", "armory 命令", attachment_type=allure.attachment_type.TEXT)
    target_options_index = kwargs.get("target_options_index", 0)
    child = wexpect.spawn("cmd.exe")
    try:
        child.expect([r">"], timeout=10)
        child.sendline(command)

        index = child.expect(["Enter to confirm."], timeout=10)
        out_aa = child.before + child.after
        if index == 0:
            if int(target_options_index) != 0:
                for char in range(int(target_options_index)):
                    child.send("\x1b[B")
                    time.sleep(0.1)  # 模拟按键间隔
            child.send("\r")  # 发送回车键
            child.expect("is active now")
            allure.attach(f"{out_aa + child.before + child.after}", "响应结果",
                          attachment_type=allure.attachment_type.JSON)
            return out_aa + child.before + child.after
    except wexpect.TIMEOUT as e:
        print('error', e)
    finally:
        child.sendline('exit')
        child.close()


def run_interactive_online_install(**kwargs):
    """
    执行远程安装时的方法
    请关闭客户端自动更新提示
    :param kwargs: 命令/选择是否同意新增文件夹
    :return:
    """
    command = kwargs.get("command")
    allure.attach(f"{command}", "armory 命令", attachment_type=allure.attachment_type.TEXT)
    choose = kwargs.get("choose")
    child = wexpect.spawn("cmd.exe")
    try:
        child.expect([r">"], timeout=10)
        child.sendline(command)

        child.expect("Download is completed!")
        aa = child.before + child.after

        index = child.expect(["/n]"], timeout=10)
        bb = child.before + child.after
        if index == 0:
            for char in choose:
                child.send(char)
                time.sleep(0.1)  # 模拟按键间隔
            child.send("\r")  # 发送回车键
            child.expect("Installation completed.")
            cc = child.before + child.after

            # 处理控制台打印内容的格式，处理了相关空行
            # 预期格式和实际操作中的格式一致
            merged_string = bb + cc
            cleaned_string = re.sub(r'\s+', ' ', merged_string).strip()
            lines = re.split(r'(?=\[)', cleaned_string)
            kk = []
            for line in lines:
                if line.strip():  # 确保不是空行
                    kk.append(line)
            ll = "\r\n".join(kk)

            allure.attach(f"{aa + ll}", "响应结果", attachment_type=allure.attachment_type.JSON)
            return aa + ll
    except wexpect.TIMEOUT:
        # child.send("\r")  # 发送回车键
        # child.expect("Installation completed.")
        print("Error,please check")
        allure.attach(f"{child.before + child.after}", "响应结果", attachment_type=allure.attachment_type.JSON)
        return child.before + child.after
    finally:
        child.sendline('exit')
        child.close()


def modify_armory_json(file_path, updates):
    """
    更新armor.json
    代码思路为-以字典的方式对整体更新，因为我提供的updates是一个全量的json格式数据
    主要考虑对应的value的类型，是dict还是list或单个值
    :param file_path: armory.json文件路径
    :param updates: 待更新的armory.json文件全量数据，格式参考armory.json文件格式,type类型为json字符串
    :return: 无返回
    """
    with open(file_path, "r", encoding="utf-8") as file:
        print('[info]初始armory.json')

        # json.load() 从文件中读取json数据，并转换为 python对象，字典或列表，实际以json的内容来决定
        data = json.load(file)
        print(data)

        # 根据 updates 字典内容（格式符合armory.json数据格式），更新data 字典中的数据
        # dict.items() 方法返回是类似于dict_items([('name', 'ylx'), ('age', 18)]
        # 使用a, b 这种写法在for循环中，可以解包
        updates = json.loads(updates)  # 处理json字符串为Python数据类型，如转换为字典

        for key, value in updates.items():
            # 仅用字典的键来匹配，若内容和字典的键匹配则 返回True，否则发货False，若你使用的这个key是一个值，那会返回False
            if key in data:

                # instance() 是python的一个内置函数，用于检查对象是否是指定的类型或其子类的实例，检查若是，则返回True,否则返回False
                if isinstance(data[key], dict) and isinstance(value, dict):
                    # 使用字典的内置方法 .update()将另一个字典的内容合并到当前字典，若目标字典中已经存在某个键则使用新值覆盖旧值
                    # 若不存在，则添加新的键值对，.前就是你的目标字典，括号内是你要提供的数据
                    data[key].update(value)
                elif isinstance(data[key], list) and isinstance(value, list):
                    # 字典更新值的方式
                    data[key] = value
                else:
                    data[key] = value
            else:
                print("---errors---非此armory.json内的字段")

        # 将修改内容写入原文件中，with open() file_path 要写入的文件路径，w是打开文件的模式，若文件已存在则覆盖写，若文件不存在则，创建一个
        # 新文件，指定文件编码为utf-8 方式
        # 使用with 打开文件的好处是，代码块执行完毕后自动关闭文件，不需要手动调用.close()方法
        # file_1 是文件对象，表示打开的文件，后续操作通过这个对象与文件交互

        with open(file_path, 'w', encoding="utf-8") as file_1:
            # json.dump() 将python的数据结构序列化为json格式，直接写入到文件中
            # json.dumps() 将数据转换为json格式的字符串，返回字符串而非写入文件
            # data 要写入的python数据结构
            # file_1 文件对象，表示json数据要写入的目标文件
            # 指定缩进级别为4个空格，生成的json文件具有良好的可读性
            # 最后一个参数确保非ASCII 字符不会被转义为Unicode编码，而是以原始字符保存
            json.dump(data, file_1, indent=4, ensure_ascii=False)

        # 读修改后的armory.json文件
        with open(file_path, "r", encoding="utf-8") as new_file:
            print('[info]更新后armory.json')
            data = json.load(new_file)
            print(data)

        print("finsh updated successfully")


def modify_readme_file(file_path, updates):
    """
    更新readme文件
    覆盖写
    :param file_path: 文件路径
    :param updates: 更新文件的内容
    :return:
    """
    with open(file_path, 'w', encoding='utf-8') as file:
        file.write(updates)

    print("finsh updated successfully")


if __name__ == "__main__":
    # 测试代码
    session = CMDSession()
    aa = session.execute("armory -v")
    print(aa)
    print("success lx")
    # session.execute('armory remote set default_remote="remote://root:root@10.12.250.46"')

    # 测试代码
    # login_info_1 = {
    #     "command": "armory login",
    #     "username": "yanglingxiao",
    #     "password": "123456"
    # }  #用户1登录信息
    #
    # login_info_2 = {
    #     "command": "armory login",
    #     "username": "yanglingxiao3311",
    #     "password": "123456"
    # }  #用户2登录信息
    #
    # publish_info_1 = {
    #     "command": "armory publish d:/@py-armory/py_armory/armory.json --access private --tag py_armory_1",
    #     "choose": "n",
    # }  #首次推送信息
    #
    # default_errors = "NO ERRORS"  #默认错误提示
    # updates_file_path_1 = r"D:\@py-armory\py_armory\armory.json"  #armory.json文件位置
    # updates_document_1 = {
    #     "name": "@py-armory/py_armory",
    #     "version": "1.0.0",
    #     "license": "",
    #     "author": "yanglingxiao3311",
    #     "description": "",
    #     "labels": [
    #         "py_armory"
    #     ],
    #     "platforms": [
    #         {
    #             "name": "windows",
    #             "arch": "x86-64",
    #             "minSupportedVersion": "",
    #             "baseDir": "",
    #             "files": ["assets"]
    #         }
    #     ],
    #     "dependencies": [],
    #     "ignore": [],
    #     "installable": False
    # }  #待更新armory.json文件
    #
    #
    #
    # session = CMDSession()
    # # armory_login_1 = run_interactive_command_login(**login_info_1)
    # # print(f"{armory_login_1}".strip())
    # # session.execute("armory whoami")
    # # output, errors = run_command("armory logout")
    # # print('output', f"{output}".strip(), "errors", errors or default_errors)
    # # armory_login_2 = run_interactive_command_login(**login_info_2)
    # # print(f"{armory_login_2}".strip())
    # # output, errors = run_command("armory whoami")
    # # print('output', f"{output}".strip(), "errors", errors or default_errors)
    # session.execute_1("cd d:/")
    # # session.execute("armory init @py-armory/py_armory_1")
    # # modify_armory_json(updates_file_path_1, updates_document_1)
    # # publish_info = run_interactive_command_publish(**publish_info_1)
    # # print(f"{publish_info}".strip())
    # session.close()
